package com.flamingskys.interview.thread;


import java.io.*;
import java.util.*;
import java.util.concurrent.*;

public class StatisticsTopKLogEvent {
    //日志文件目录
    public static final String baseDir = "/Users/flamingskys/Downloads/";

    public static void main(String[] args) {
        try {
            completionServiceCount();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
    /**
     * 使用completionService收集callable结果，输出所有线程的汇总统计结果
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void completionServiceCount() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        CompletionService<Map<String, Integer>> completionService = new ExecutorCompletionService<>(
                executorService);
        int threadNum = 5;
        for (int i = 1; i <= threadNum; i++) {
            completionService.submit(getTask(i));
        }
        Map<String, Integer> resultMap = new HashMap<>();
        for(int i=1;i <= threadNum;i++){
            Map<String, Integer> map = completionService.take().get();

            //合并map
            Map<String, Integer> lastMap = new HashMap<String, Integer>(resultMap);
            map.forEach((key, value) -> lastMap.merge(key, value, (v1, v2) -> (v1+v2)));
            resultMap = new HashMap<String, Integer>(lastMap);;
        }

        List<OpEventItem> eventItemList = new ArrayList<>();
        for(String key:resultMap.keySet()){
            eventItemList.add(new OpEventItem(key, resultMap.get(key)));
        }
        //返回出现次数最多的10种错误类型及其个数
        int TopK = 10;
        PriorityQueue<OpEventItem> maxHeap = new PriorityQueue<OpEventItem>(TopK, new Comparator<OpEventItem>() {
            public int compare(OpEventItem o1, OpEventItem o2) {
                return o2.count-o1.count;
            }
        });

        for(OpEventItem item : eventItemList){
            maxHeap.offer(item);
        }

        for (int i = 0; i < TopK; i++) {
            OpEventItem opEventItem1 = maxHeap.poll();
            System.out.println(opEventItem1.type+":"+opEventItem1.count);
        }

        executorService.shutdown();
    }

    /**
     * 读取文件内容并分割，统计单个文件的各事件次数
     * @param no
     * @return
     */
    public static Callable<Map<String, Integer>> getTask(final int no) {
        Callable<Map<String, Integer>> task = new Callable<Map<String, Integer>>() {
            @Override
            public Map<String, Integer> call() throws Exception {
                String filePath = baseDir+"filePath"+no;
                File readFile = new File(filePath);

                Map<String, Integer> countMap = new HashMap<>();

                //输入IO流声明
                InputStream in = null;
                InputStreamReader ir = null;
                BufferedReader br = null;

                try {
                    in = new BufferedInputStream(new FileInputStream(readFile));
                    //暂不考虑编码问题
                    ir = new InputStreamReader(in);
                    br = new BufferedReader(ir);
                    String line = "";

                    //一行一行读取
                    while ((line = br.readLine()) != null) {
                        String[] itemArr = line.split(":");
                        String type = itemArr[0];
                        if(!countMap.containsKey(type)){
                            countMap.put(type, 1);
                        }else{
                            countMap.put(type, countMap.get(type)+1);
                        }
                    }
                }
                catch (Exception e) {
                    e.printStackTrace();
                }
                finally{
                    try {
                        if(br!=null){
                            br.close();
                        }
                        if(ir!=null){
                            ir.close();
                        }
                        if(in!=null){
                            in.close();
                        }
                    } catch (Exception e2) {

                    }
                }
                return countMap;
            }
        };
        return task;
    }
}

class OpEventItem{
    String type;
    int count;

    public OpEventItem(String type, int count){
        this.type = type;
        this.count = count;
    }
}